package cn.com.pc.content.service.es.impl;

import cn.com.pc.content.service.es.ESCurdService;
import cn.com.pc.content.service.es.ElasticEntity;
import cn.com.pc.content.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.function.Function;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

/**
 * @description es的基础增删改查操作实现类
 */
@Slf4j
@Service
@Primary
public class ESCurdServiceImpl implements ESCurdService {

    private RestHighLevelClient highLevelClient;

    @Resource
    public void setHighLevelClient(RestHighLevelClient highLevelClient) {
        this.highLevelClient = highLevelClient;
    }

    /**
     * 功能描述: 创建索引
     *
     * @param indexName 索引名称
     * @return:void
     */
    @Override
    public String createIndex(String indexName) throws IOException {
        // 创建索引
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
        createIndexRequest.settings(Settings.builder()
                // 分片数
                .put("index.number_of_shards", 3)
                // 备份数
                .put("index.number_of_replicas", 2));
        CreateIndexResponse response = highLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        return response.index();
    }

    /**
     * 创建索引
     *
     * @param idxName  索引名称
     * @param settings 索引配置信息
     * @param mappings map映射信息
     */
    @Override
    public void createIndex(String idxName, Map<String, Object> settings, Map<String, Object> mappings) {
        try {
            if (!this.existIndex(idxName)) {
                log.error(" idxName={} 已经存在", idxName);
                return;
            }
            CreateIndexRequest request = new CreateIndexRequest(idxName);
//            if (StringUtils.isNotEmpty(settings)) {
//                request.settings(settings, XContentType.JSON);
//            }
//            if (StringUtils.isNotEmpty(mappings)) {
//                request.mapping(mappings, XContentType.JSON);
//            }
            if (settings != null && !settings.isEmpty()) {
                request.settings(settings);
            }
            if (mappings != null && !mappings.isEmpty()) {
                request.mapping(mappings);
            }
            CreateIndexResponse response = highLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!response.isAcknowledged()) {
                throw new RuntimeException("初始化失败！");
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(0);
        }
    }

    @Override
    public void bulkOp(BulkRequest bulkRequest) {
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);// 设置刷新策略
        bulkRequest.timeout(TimeValue.timeValueMinutes(30));//设置超时时间
//        bulkRequest.waitForActiveShards(2); //设置在继续索引/更新/删除操作之前必须活跃的碎片副本的数量
//        bulkRequest.pipeline("pipline_id");//用于所有子请求的全局管道标识

        highLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                log.info("批量操作成功");
                for (BulkItemResponse itemResponse : bulkItemResponses) {
                    // 判断给定操作是否失败
                    if (itemResponse.isFailed()) {
                        log.error(itemResponse.getFailureMessage());
                        // 检索失败后操作
//                        BulkItemResponse.Failure failure = itemResponse.getFailure();
                    } else {
                        DocWriteResponse writeResponse = itemResponse.getResponse();
//                        switch (itemResponse.getOpType()) {
//                            case INDEX:
////                            IndexResponse indexResponse = (IndexResponse)writeResponse;
//                                break;
//                            case CREATE:
//                                break;
//                            case DELETE:
//                                break;
//                            case UPDATE:
//                                break;
//                        }
                    }
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage());
            }
        });


    }

    /**
     * 功能描述: 删除索引
     *
     * @param indexName 索引名称
     * @return:void
     */
    @Override
    public void deleteIndex(String indexName) throws IOException {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest();
        deleteIndexRequest.indices(indexName);
        highLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
    }

    /**
     * 功能描述: 查询索引
     *
     * @param indexName 索引名称
     * @return:void
     */
    @Override
    public boolean existIndex(String indexName) throws IOException {
        // 判断索引是否存在
        GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
        return highLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
    }

    /**
     * 功能描述: 往索引里插入数据
     *
     * @param indexName 索引名称
     * @param docType   数据对象类型，可以把它设置成对象类的类名
     * @param object    数据对象
     * @return:void
     */
    @Override
    public RestStatus insert(String indexName, String docType, Object object) throws IOException {
        IndexRequest indexRequest = new IndexRequest();
        // source方法里面需要填写对应数据的类型，默认数据类型为json
        indexRequest.index(indexName).source(JsonUtils.obj2String(object), XContentType.JSON);
        IndexResponse response = highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        return response.status();
    }

    @Override
    public void insertAsync(String indexName, String docType, Object object) {
        IndexRequest indexRequest = new IndexRequest();
        // source方法里面需要填写对应数据的类型，默认数据类型为json
        indexRequest.index(indexName).source(JsonUtils.obj2String(object), XContentType.JSON);
        highLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                if (indexResponse.status() != RestStatus.CREATED) {
                    log.error("index: {}, id: {}, 插入数据失败", indexResponse.getIndex(), indexResponse.getId());
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage(), e);
            }
        });
    }

    /**
     * 批量插入数据
     *
     * @param idxName 索引名称
     * @param list    待插入列表数据
     */
    @Override
    public void insertBatch(String idxName, List<ElasticEntity> list) {
        BulkRequest request = new BulkRequest();
        list.forEach(item -> request.add(new IndexRequest(idxName).id(item.getId())
                .source(JsonUtils.obj2String(item.getData()), XContentType.JSON)));
        highLevelClient.bulkAsync(request, RequestOptions.DEFAULT, actionListener(idxName, list.size()));
    }

    /**
     * 功能描述: 更新数据
     *
     * @param indexName 索引名称
     * @param object    数据对象
     * @return:void
     */
    @Override
    public RestStatus update(String indexName, String id, Object object) throws IOException {
        UpdateRequest request = new UpdateRequest();
        request.index(indexName).id(id).doc(JsonUtils.obj2String(object), XContentType.JSON);
        UpdateResponse response = highLevelClient.update(request, RequestOptions.DEFAULT);
        return response.status();
    }

    @Override
    public void updateAsync(String indexName, String id, Object object) {
        UpdateRequest request = new UpdateRequest();
        request.index(indexName).id(id).doc(JsonUtils.obj2String(object), XContentType.JSON);
        highLevelClient.updateAsync(request, RequestOptions.DEFAULT, new ActionListener<UpdateResponse>() {
            @Override
            public void onResponse(UpdateResponse updateResponse) {
                if (updateResponse.status() != RestStatus.OK) {
                    log.error("index: {}, id: {}, 修改数据失败", updateResponse.getIndex(), updateResponse.getId());
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage(), e);
            }
        });
    }

    /**
     * 根据 id 更新指定索引中的文档
     */
    @Override
    public RestStatus updateDoc(String idxName, String id, String updateJson) throws IOException {
        UpdateRequest request = new UpdateRequest(idxName, id);
        request.doc(XContentType.JSON, updateJson);
        UpdateResponse response = highLevelClient.update(request, RequestOptions.DEFAULT);
        return response.status();
    }

    /**
     * 批量更新数据
     *
     * @param idxName 索引名称
     * @param list    待插入列表数据
     */
    @Override
    public void updateBatch(String idxName, List<ElasticEntity> list) {
        BulkRequest request = new BulkRequest();
        list.forEach(item -> request.add(new UpdateRequest(idxName, item.getId())
                .doc(item.getData()).docAsUpsert(true)));
//        BulkResponse response = highLevelClient.bulk(request, RequestOptions.DEFAULT);
        highLevelClient.bulkAsync(request.timeout(TimeValue.timeValueHours(1)),
                RequestOptions.DEFAULT, actionListener(idxName, list.size()));
    }

    /**
     * 根据查询条件更新索引中的文档
     */
    @Override
    public void updateByQuery(String idxName, QueryBuilder queryBuilder, Map<String, Object> updateData) {
        UpdateByQueryRequest request = new UpdateByQueryRequest(idxName);
        //默认情况下，版本冲突会中止UpdateByQueryRequest进程，
//        request.setConflicts("proceed"); //设置版本冲突时继续
        request.setAbortOnVersionConflict(false);//设置版本冲突时继续
        request.setQuery(queryBuilder);
        // 设置大小来限制已处理文档的数量 request.setSize(100);
        // 默认情况下单次可以处理1000个文档，可以更改批处理大小
        request.setBatchSize(100);
//        request.setPipeline("my_pipline"); // 设置管道线
//        request.setSlices(2); //设置使用的分片数
//        request.setRouting("=car"); // 设置路由
        // 使用scroll参数来控制它将“搜索上下文”保持多长时间
        request.setScroll(TimeValue.timeValueMinutes(5));//设置滚动时间
        request.setRefresh(true);

        if (updateData == null || updateData.isEmpty()) {
            return;
        }
        StringBuilder sb = new StringBuilder();
        int size = 0;
        for (Map.Entry<String, Object> entry : updateData.entrySet()) {
            sb.append("ctx._source['").append(entry.getKey()).append("'] = ");
            Object value = entry.getValue();
            if (value instanceof Integer
                    || value instanceof Double || value instanceof Float
                    || value instanceof Boolean) {
                sb.append(entry.getValue());
            } else if (value instanceof List) {
                sb.append(JsonUtils.obj2String(value));
            } else if (value instanceof Long) {
                sb.append(entry.getValue()).append('L');
            } else if (value instanceof Date) {
                sb.append("'").append(((Date) value).getTime()).append("'");
            } else {
                sb.append("'").append(entry.getValue()).append("'");
            }
            size++;
            if (size < updateData.size()) {
                sb.append(";");
            }
        }
        Script script = new Script(sb.toString());
        log.info(sb.toString());
        request.setScript(script);
        // 同步执行
//        BulkByScrollResponse response = highLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
        // 异步执行
        highLevelClient.updateByQueryAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
            @Override
            public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
                log.info("Total: " + bulkByScrollResponse.getTotal() + ", Updated:" + bulkByScrollResponse.getUpdated());
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage(), e);
            }
        });
    }

    /**
     * 根据查询条件对大数据逐条执行业务
     */
    @Override
    public void updateSearchScroll(SearchSourceBuilder searchSourceBuilder, String idxName, Function<SearchHit, Object> func) throws IOException {
        Scroll scroll = new Scroll(TimeValue.timeValueHours(1));
        SearchRequest request = new SearchRequest(idxName).source(searchSourceBuilder).scroll(scroll);
        SearchResponse searchResponse = highLevelClient.search(request, RequestOptions.DEFAULT);
        ;
        String scrollId = searchResponse.getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        long totalHits = searchResponse.getHits().getTotalHits().value;
        log.info("Total:{}", totalHits);
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        int i = 0;
        while (searchHits != null && searchHits.length > 0) {
            if (StringUtils.hasLength(scrollId)) {
                clearScrollRequest.addScrollId(scrollId);
            }
            for (SearchHit hit : searchResponse.getHits().getHits()) {
                func.apply(hit);
                i++;
            }
            searchResponse = highLevelClient.scroll(new SearchScrollRequest(scrollId).scroll(scroll), RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
            log.info("Finish {}%", i * 100 / totalHits);
        }
        highLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
    }

    /**
     * 索引之间导数据
     */
    @Override
    public void exportData(HashMap<String, String> fieldMap, String fromIndex, String toIndex) {
        List<ElasticEntity> elasticEntities = null;
        Object[] objects = new Object[]{};
        ElasticEntity elasticEntity;
        try {
            SearchSourceBuilder builder = new SearchSourceBuilder()
                    .query(QueryBuilders.matchAllQuery()).timeout(TimeValue.timeValueMinutes(30))
                    .fetchSource(fieldMap.values().toArray(new String[0]),
                            new String[]{});
            SearchRequest searchRequest = new SearchRequest(fromIndex);
            builder.size(1000);
            builder.sort("id", SortOrder.ASC);

            while (true) {
                elasticEntities = new ArrayList<>();

                if (objects.length > 0) {
                    builder.searchAfter(objects);
                }
                SearchResponse searchResponse = highLevelClient.search(searchRequest.source(builder), RequestOptions.DEFAULT);
                SearchHit[] hits = searchResponse.getHits().getHits();
                if (hits.length == 0) {
                    break;
                }
                objects = hits[hits.length - 1].getSortValues();
                for (SearchHit hit : hits) {
                    Map<String, Object> data = hit.getSourceAsMap();
                    Map<String, Object> map = new HashMap<>();
                    Integer id = data.values().toString().hashCode();
                    for (String key : fieldMap.keySet()) {
                        if ("id".equals(key)) {
                            map.put("id", id);
                        }
                        map.put(key, data.get(fieldMap.get(key)));
                    }

                    elasticEntity = new ElasticEntity();
                    elasticEntity.setId(String.valueOf(id));
                    elasticEntity.setData(map);
                    elasticEntities.add(elasticEntity);
                }
                try {
                    updateBatch(toIndex, elasticEntities);
                } catch (Exception e) {
                    log.error("updateBatch: " + toIndex + ",Exception:" + e.getMessage());
                }

            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        log.info("searchScrollAfter: " + elasticEntities.size());
    }

    /**
     * 功能描述: 根据id删除数据
     *
     * @param id _id编号
     * @return:org.elasticsearch.rest.RestStatus
     */
    @Override
    public RestStatus delete(String id) throws IOException {
        DeleteRequest request = new DeleteRequest();
        request.id(id);
        DeleteResponse response = highLevelClient.delete(request, RequestOptions.DEFAULT);
        return response.status();
    }

    /**
     * 根据 id 删除指定索引中的文档
     */
    @Override
    public RestStatus deleteDoc(String idxName, String id) throws IOException {
        DeleteRequest request = new DeleteRequest(idxName, id);
        DeleteResponse response = highLevelClient.delete(request, RequestOptions.DEFAULT);
        return response.status();
    }

    /**
     * 批量删除
     */
    @Override
    public <T> void deleteBatch(String idxName, Collection<T> idList) {
        BulkRequest request = new BulkRequest();
        idList.forEach(item -> request.add(new DeleteRequest(idxName, item.toString())));
        try {
            highLevelClient.bulk(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量查询删除
     */
    @Override
    public void deleteByQuery(String idxName, QueryBuilder queryBuilder) {
        DeleteByQueryRequest request = new DeleteByQueryRequest(idxName);
        //默认情况下，版本冲突会中止DeletedByQueryRequest进程，
//        request.setConflicts("proceed"); //设置版本冲突时继续
        request.setAbortOnVersionConflict(false);//设置版本冲突时继续
        request.setQuery(queryBuilder);
        // 设置大小来限制已处理文档的数量 request.setSize(100);
        // 默认情况下单次可以处理1000个文档，可以更改批处理大小
        request.setBatchSize(100);
//        request.setSlices(2); //设置使用的分片数
//        request.setRouting("=car"); // 设置路由
        // 使用scroll参数来控制它将“搜索上下文”保持多长时间
        request.setScroll(TimeValue.timeValueMinutes(5));//设置滚动时间
        request.setRefresh(true);

        // 同步执行
//        BulkByScrollResponse response = highLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
        // 异步执行
        highLevelClient.deleteByQueryAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
            @Override
            public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
                log.info("Total: " + bulkByScrollResponse.getTotal() + ", Deleted:" + bulkByScrollResponse.getDeleted());
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage(), e);
            }
        });

    }

    /**
     * 批量操作结果统一处理
     * 失败日志
     */
    private ActionListener actionListener(String idxName, int total) {
        return new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                int i = 0;
                Map<String, Integer> mapCount = new HashMap<>();
                for (BulkItemResponse itemResponse : bulkItemResponses) {
                    if (itemResponse.isFailed()) {
                        i++;
                        log.error("id: " + itemResponse.getId() + ", OpType: " + itemResponse.getOpType()
                                + " ,error: " + itemResponse.getFailureMessage());
//                            BulkItemResponse.Failure failure = itemResponse.getFailure();
                    } else {
                        String op = itemResponse.getOpType().getLowercase();
                        mapCount.put(op, mapCount.getOrDefault(op, 0) + 1);
                    }
                }
                StringBuilder sb = new StringBuilder();
                mapCount.keySet().forEach((k) -> {
                    sb.append(k);
                    sb.append(":");
                    sb.append(mapCount.get(k));
                    sb.append("; ");
                });
                log.info("bulk: " + idxName + ", " + sb.toString() + ", fail:" + i + "/" + total);
            }

            @Override
            public void onFailure(Exception e) {
                log.error(e.getMessage());
            }
        };
    }

    /**
     * 功能描述: 查询索引是否存在
     *
     * @param indexName 索引名称
     * @param id        索引id
     * @return
     * @throws IOException
     */
    @Override
    public boolean existDocById(String indexName, String id) throws IOException {
        // 判断索引是否存在
        GetRequest getRequest = new GetRequest(indexName, id);
        return highLevelClient.exists(getRequest, RequestOptions.DEFAULT);
    }

    /**
     * 功能描述: 往索引里插入数据，并指定Id
     *
     * @param indexName 索引名称
     * @param object    数据对象
     * @return:void
     */
    @Override
    public RestStatus insertWithId(String indexName, String id, Object object) throws IOException {
        IndexRequest indexRequest = new IndexRequest();
        // source方法里面需要填写对应数据的类型，默认数据类型为json
        indexRequest.index(indexName).id(id).source(JsonUtils.obj2String(object), XContentType.JSON);
        IndexResponse response = highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        return response.status();
    }

    /**
     * 功能描述: 往索引里添加新字段
     *
     * @param indexName
     * @param propName
     * @param propType
     * @throws IOException
     */
    @Override
    public boolean insertMappingProperty(String indexName, String propName, String propType) throws IOException {
        PutMappingRequest putMappingRequest = new PutMappingRequest(indexName);
        XContentBuilder builder;
        if (propType.equals("date")) {
            builder = jsonBuilder().startObject()
                    .startObject("properties")
                    .startObject("metaData")
                    .startObject("properties")
                    .startObject(propName)
                    .field("type", propType)
                    .field("format", "yyyy-MM-dd HH:mm:ss||epoch_millis")
                    .endObject()
                    .endObject()
                    .endObject()
                    .endObject()
                    .endObject();
        } else {
            builder = jsonBuilder().startObject()
                    .startObject("properties")
                    .startObject("metaData")
                    .startObject("properties")
                    .startObject(propName)
                    .field("type", propType)
                    .endObject()
                    .endObject()
                    .endObject()
                    .endObject()
                    .endObject();
        }
        putMappingRequest.source(builder);
        AcknowledgedResponse acknowledgedResponse = highLevelClient.indices().putMapping(putMappingRequest, RequestOptions.DEFAULT);
        return acknowledgedResponse.isAcknowledged();
    }
}
